package src.main.scala.com.lagou.homework.part05

import java.util.UUID

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
import scala.concurrent.duration._

/**
 * @author: yehw
 * @date: 2020/10/11 14:18
 * @description: worker端
 */
//Worker端
class Worker(val memory: Int, val cores: Int, val masterHost: String, val masterPort: String) extends Actor {

  //定义一个workerId
  private val workerId: String = UUID.randomUUID().toString
  //定义常量 表示每隔多久向master发送一次心跳
  val sendHeartBeatInterval = 10000 //10秒
  //定义master全局变量
  var master: ActorSelection = _
  //preStart方法会在构造代码执行后被调用，并且只被运行一次，可以在这里做一些初始化的工作
  override def preStart(): Unit = {
    //获取master引用
    //通过ActorContext上下文对象调用actorSelection可以从已知的actor中寻找目标
    //方法中需要一个字符串,它包括了构建master actor的条件
    //1、通信协议 2、master的ip 3、master端口  4、创建master actor老大 5、master actor 名称  6、actor层级关系
    master = context.actorSelection(s"akka.tcp://masterActorSystem@$masterHost:$masterPort/user/masterActor")

    //通过master引用向master发送注册信息，通过样例类封装注册信息
    master ! RegisterMessage(workerId, memory, cores)

  }

  //不断接受消息
  override def receive: Receive = {
    //worker接受master反馈的注册成功信息
    case RegisteredMessage(message) => {
      println(message)

      //定时向master发送心跳
      //正常是需要向master引用发送心跳，由于这里类型不符合，就需要使用self---->表示worker actor本身

      //手动导入隐式转换
      import context.dispatcher
      context.system.scheduler.schedule(0 millis, sendHeartBeatInterval millis, self, SendHeartBeat)

    }
    //worker自己向自己发送信息
    case SendHeartBeat => {
      //worker真正向master发送心跳
      master ! HeartBeat(workerId)
    }
  }
}

object Worker {
  def main(args: Array[String]): Unit = {
    //定义worker的IP地址
    val host = "127.0.0.1"
    //定义worker的端口
    val port = "8081"

    //定义worker的内存大小
    val memory = 20
    //定义worker的cpu核数
    val cores = 3

    //定义master的IP地址
    val masterHost ="127.0.0.1"
    //定义master的端口
    val masterPort = "8080"

    //准备字符串配置信息
    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
        """.stripMargin

    //通过ConfigFactory调用parseString解析字符串获取config对象
    val config: Config = ConfigFactory.parseString(configStr)

    //1、创建ActorSystem 负责创建和监督actor
    val workerActorSystem = ActorSystem("workerActorSystem", config)

    //2、创建worker actor
    val workerActor: ActorRef = workerActorSystem.actorOf(Props(new Worker(memory, cores, masterHost, masterPort)), "workerActor")

    //3、测试
    //workerActor ! "connect"
  }
}




